/*
 * similarity_detection.c
 *
 *  Created on: Mar 25, 2014
 *      Author: fumin
 */
#include "index.h"
#include "kvstore.h"
#include "fingerprint_cache.h"
#include "../recipe/recipestore.h"
#include "../storage/containerstore.h"
#include "../jcr.h"

/* defined in index.c */
extern struct
{
    /* map a fingerprint to a queue of indexElem */
    /* Index all fingerprints in the index buffer. */
    GHashTable *buffered_fingerprints;
    /* The number of buffered chunks */
    int num;
} index_buffer;

/* defined in index.c */
extern struct
{
    /* Requests to the key-value store */
    int lookup_requests;
    int update_requests;
    int lookup_requests_for_unique;
    /* Overheads of prefetching module */
    int read_prefetching_units;
} index_overhead;

/*
 * Larger one comes before smaller one.
 * Descending order.
 */
static gint g_segment_cmp_feature_num(struct segment *a,
                                      struct segment *b, gpointer user_data)
{
    gint ret = g_hash_table_size(b->features) - g_hash_table_size(a->features);
    if (ret == 0)
    {
        ret = b->id > a->id ? 1 : -1;
        return ret;
    }
    else
        return ret;
}

/*
 * Remove the features that are common with top from target.
 */
static void features_trim(struct segment *target,
                          struct segment *top)
{
    GHashTableIter iter;
    gpointer key, value;
    g_hash_table_iter_init(&iter, top->features);
    while (g_hash_table_iter_next(&iter, &key, &value))
    {
        g_hash_table_remove(target->features, key);
    }
}

/*
 * Select the top segments that are most similar with features.
 * (top-k * prefetching_num) cannot be larger than the segment cache size.
 */
static void top_segment_select(GHashTable *features)
{
    /*
	 * Mapping segment IDs to similar segments that hold at least one of features.
	 */
    GHashTable *similar_segments = g_hash_table_new_full(g_int64_hash, g_int64_equal, NULL,
                                                         free_segment);

    GHashTableIter iter;
    gpointer key, value;
    g_hash_table_iter_init(&iter, features);
    /* Iterate the features of the segment. */
    while (g_hash_table_iter_next(&iter, &key, &value))
    {
        /* Each feature is mapped to several segment IDs. */
        segmentid *ids = kvstore_lookup((fingerprint *)key);
        if (ids)
        {
            index_overhead.lookup_requests++;
            int i;
            for (i = 0; i < destor.index_value_length; i++)
            {
                if (ids[i] == TEMPORARY_ID)
                    break;
                struct segment *s = g_hash_table_lookup(similar_segments, &ids[i]);
                if (!s)
                {
                    s = new_segment_full();
                    s->id = ids[i];
                    g_hash_table_insert(similar_segments, &s->id, s);
                }
                char *feature = malloc(destor.index_key_size);
                memcpy(feature, key, destor.index_key_size);
                assert(!g_hash_table_contains(s->features, feature));
                g_hash_table_insert(s->features, feature, NULL);
            }
        }
        else
        {
            index_overhead.lookup_requests_for_unique++;
        }
    }

    if (g_hash_table_size(similar_segments) != 0)
    {

        /* Sorting similar segments in order of their number of hit features. */
        GSequence *seq = g_sequence_new(NULL);
        GHashTableIter iter;
        gpointer key, value;
        g_hash_table_iter_init(&iter, similar_segments);
        while (g_hash_table_iter_next(&iter, &key, &value))
        {
            /* Insert similar segments into GSequence. */
            struct segment *s = value;
            //NOTICE("candidate segment %lld with %d shared features", s->id, g_hash_table_size(s->features));
            g_sequence_insert_sorted(seq, s, g_segment_cmp_feature_num, NULL);
        }

        /* The number of selected similar segments */
        int num = g_sequence_get_length(seq) > destor.index_segment_selection_method[1] ? destor.index_segment_selection_method[1] : g_sequence_get_length(seq), i;

        //NOTICE("select Top-%d in %d segments", num, g_sequence_get_length(seq));

        /* Prefetched top similar segments are pushed into the queue. */
        for (i = 0; i < num; i++)
        {
            /* Get the top segment */
            struct segment *top = g_sequence_get(
                g_sequence_get_begin_iter(seq));
            //NOTICE("read segment %lld", top->id);

            index_overhead.read_prefetching_units++;
            fingerprint_cache_prefetch(top->id);

            g_sequence_remove(g_sequence_get_begin_iter(seq));
            g_sequence_foreach(seq, features_trim, top);
            g_sequence_sort(seq, g_segment_cmp_feature_num, NULL);
        }
        g_sequence_free(seq);
    }

    g_hash_table_destroy(similar_segments);
}

extern struct
{
    /* accessed in dedup phase */
    struct container *container_buffer;
    /* In order to facilitate sampling in container,
	 * we keep a queue for chunks in container buffer. */
    GSequence *chunks;
} storage_buffer;

void index_lookup_similarity_detection(struct segment *s)
{
    assert(s->features);
    top_segment_select(s->features);

    int len = g_queue_get_length(s->chunks), i;
    for (i = 0; i < len; ++i)
    {
        struct chunk *c = g_queue_peek_nth(s->chunks, i);

        if (CHECK_CHUNK(c, CHUNK_FILE_START) || CHECK_CHUNK(c, CHUNK_FILE_END))
            continue;

        /* First check it in the storage buffer */
        if (storage_buffer.container_buffer && lookup_chunk_in_container(storage_buffer.container_buffer, &c->fp))
        {
            c->id = get_container_id(storage_buffer.container_buffer);
            SET_CHUNK(c, CHUNK_DUPLICATE);
            SET_CHUNK(c, CHUNK_REWRITE_DENIED);
        }
        /*
		 * First check the buffered fingerprints,
		 * recently backup fingerprints.
		 */
        GQueue *tq = g_hash_table_lookup(index_buffer.buffered_fingerprints, &c->fp);
        if (!tq)
        {
            tq = g_queue_new();
        }
        else if (!CHECK_CHUNK(c, CHUNK_DUPLICATE))
        {
            /* chunk exist but is not duplicate */
            struct indexElem *be = g_queue_peek_head(tq);
            c->id = be->id;
            SET_CHUNK(c, CHUNK_DUPLICATE);
            if (c->id == TEMPORARY_ID)
                c->type = CHUNK_TYPE_DUPLICATE_ALL; // The buffered fingerprints have not the segment recipes
            else
            {
                if (be->type == CHUNK_TYPE_UNIQUE || be->type == CHUNK_TYPE_DUPLICATE)
                    c->type = CHUNK_TYPE_DUPLICATE;
                else
                    c->type = CHUNK_TYPE_DUPDELTA;
            }
        }

        /* Check the fingerprint cache */
        if (!CHECK_CHUNK(c, CHUNK_DUPLICATE))
        {
            /* Searching in fingerprint cache */
            int64_t id = delta_fingerprint_cache_lookup(c);
            if (id != TEMPORARY_ID)
            {
                c->id = id;
                SET_CHUNK(c, CHUNK_DUPLICATE);
            }
        }

        if (!CHECK_CHUNK(c, CHUNK_DUPLICATE))
        {
            /*		if (in_dict(filter, (char*)&c->fp, sizeof(fingerprint))) {
                containerid cid = db_lookup_fingerprint(&c->fp);
                    
                if (cid != TEMPORARY_ID) {
                    index_overhead.lookup_requests++;
                    
                    index_overhead.read_prefetching_units++;

                    fingerprint_cache_prefetch(cid);
                        
                    int64_t id = delta_fingerprint_cache_lookup(c);
                    if(id != TEMPORARY_ID){
                        c->id = id;
                        SET_CHUNK(c, CHUNK_DUPLICATE);
                    }
                    else{
                        NOTICE("Filter phase: A key collision occurs");
                    }
                }
            }
    */
            int64_t *ids = kvstore_lookup((char *)&c->fp);
            if (ids)
            {
                fingerprint_cache_prefetch(ids[0]);
                int64_t id = delta_fingerprint_cache_lookup(c);
                if (id != TEMPORARY_ID)
                {
                    c->id = id;
                    SET_CHUNK(c, CHUNK_DUPLICATE);
                }
                else
                {
                    NOTICE("Filter phase: A key collision occurs");
                }
            }
            else
            {
                index_overhead.lookup_requests_for_unique++;
                VERBOSE("Dedup phase: non-existing fingerprint");
            }
        }

        if (!CHECK_CHUNK(c, CHUNK_DUPLICATE))
        {
            c->type = CHUNK_TYPE_UNIQUE_ALL;
        }

        /* Insert it into the index buffer */
        struct indexElem *ne = (struct indexElem *)malloc(
            sizeof(struct indexElem));
        ne->id = c->id;
        ne->type = c->type;
        memcpy(&ne->fp, &c->fp, sizeof(fingerprint));

        g_queue_push_tail(tq, ne);
        g_hash_table_replace(index_buffer.buffered_fingerprints, &ne->fp, tq);

        index_buffer.num++;
    }
}
